package com.aliyun.messagemgt.loghub;

import com.aliyun.openservices.log.common.FastLog;
import com.aliyun.openservices.log.common.FastLogContent;
import com.aliyun.openservices.log.common.FastLogGroup;
import com.aliyun.openservices.log.common.LogGroupData;
import com.aliyun.openservices.loghub.client.ILogHubCheckPointTracker;
import com.aliyun.openservices.loghub.client.exceptions.LogHubCheckPointException;
import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author changcong
 */
@Component
public class LogHubProcessor implements ILogHubProcessor {

    private static Logger logger = LoggerFactory.getLogger(LogHubProcessor.class);

    private static final int TIMEUNIT = 1000;

    private static final int TIME = 30;

    private int mShardId;

    private long mLastCheckTime = 0;

    @Override
    public void initialize(int shardId) {
        this.mShardId = shardId;
    }

    /**
     * 消费数据的主逻辑，这里面的所有异常都需要捕获，不能抛出去。
     */
    @Override
    public String process(List<LogGroupData> logGroups,
                          ILogHubCheckPointTracker checkPointTracker) {
        List<Map<String, Object>> loggers = new ArrayList<>();
        FastLogGroup flg = logGroups.get(0).GetFastLogGroup();
        for (int lIdx = 0; lIdx < flg.getLogsCount(); ++lIdx) {
            FastLog log = flg.getLogs(lIdx);
            // Key的list数组
            List listKey = new ArrayList(10);
            for (int cIdx = 0; cIdx < log.getContentsCount(); ++cIdx) {
                FastLogContent content = log.getContents(cIdx);
                listKey.add(content.getKey());
            }
        }
        return null;
    }

    /**
     * 当 worker 退出的时候，会调用该函数，用户可以在此处做些清理工作。
     */
    @Override
    public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
        //将消费断点保存到服务端。
        try {
            checkPointTracker.saveCheckPoint(true);
        } catch (LogHubCheckPointException e) {
           logger.error("shutdown LogHubCheckPointException:",e);
        }
    }

}
